package ca.cbc.msb.mq;

import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.naming.InitialContext;
import javax.naming.NamingException;

import org.hornetq.jms.client.HornetQConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import ca.cbc.msb.entity.MSBMessage;
import ca.cbc.msb.enums.QueueEnum;

/**
 * JMS implementation of MessageUtil
 * 
 * @author usr
 *
 */
public class MessageUtilImpl implements MessageUtil {

	private Logger log = LoggerFactory.getLogger(this.getClass());
	private Connection jmsConnection;
	private final boolean transacted = true;
	//message live-time: 250 minutes
	private final static long MESSAGE_LIVE_TIME = 1000 * 60 * 250; 
	
//	 private final ExceptionListener exceptionListener = new ExceptionListener()
//	   {
//	      public void onException(final JMSException e)
//	      {
//	    	  log.error("!!! MQ server connection error, reconnecting...");
//	         disconnect();
//	         try {
//				Thread.sleep(10000);
//			} catch (InterruptedException e1) {
//				log.error("!!!Sleep interrupted: ",e1);
//			}
//	         try {
//				connect();
//			} catch (NamingException e1) {
//				log.error("!!!JNDI finding server error: ",e);
//			} catch (JMSException e1) {
//				log.error("!!!reconnect to MQ server error: ",e);
//			}
//	      }
//	   };
	   private Session session;
	   private Map<QueueEnum,MessageProducer > pMap = new HashMap<QueueEnum,MessageProducer >();
	   private Map<QueueEnum,MessageConsumer > cMap = new HashMap<QueueEnum,MessageConsumer >();
	   private Map<QueueEnum,Destination > dMap = new HashMap<QueueEnum,Destination >();
	   private  InitialContext ic;
	   
	   
	   public MessageUtilImpl() throws NamingException, JMSException {
		   connect();
	   }
	   
	   private void connect() throws NamingException, JMSException {	      
	       ic = new InitialContext();
           log.debug("initial context created");
	       HornetQConnectionFactory factory = (HornetQConnectionFactory)ic.lookup("/ConnectionFactory");
           factory.setRetryInterval(2000);
           factory.setReconnectAttempts(1000);
           log.debug("creating jms connection");
	         jmsConnection = factory.createConnection();
//	         jmsConnection.setExceptionListener(exceptionListener);
	         session=jmsConnection.createSession(transacted,  Session.SESSION_TRANSACTED);//AUTO_ACKNOWLEDGE);
	         jmsConnection.start();
	   }
	   
	   public void disconnect()
	   {
		log.debug("Disconnecting...");
	      if (jmsConnection != null)
	      {
	         try
	         {
	        	 session.close();
	        	 jmsConnection.setExceptionListener(null);
	        	 jmsConnection.close();
	         }
	         catch (JMSException e)
	         {
	           log.error("!!!Error in closing session in MessageUtil",e);
	         }
	         finally
	         {
	        	 jmsConnection = null;
	         }
	      }
	   }
	   
	   public MSBMessage getMessage(QueueEnum queueName) throws NamingException, JMSException{

			ObjectMessage objMessage =null;
			Destination destination=dMap.get(queueName);
			if (destination==null){
			 destination = (Destination)ic.lookup("/queue/"+queueName.toString());
			 dMap.put(queueName, destination);
			}
			MessageConsumer consumer = cMap.get(queueName);
			if(consumer==null){
				consumer = session.createConsumer(destination);
				cMap.put(queueName, consumer);
			}

			objMessage = (ObjectMessage) consumer.receive(10000);
			if(objMessage == null){
				return null;
			}

			objMessage.acknowledge();
			MSBMessage msg= (MSBMessage) objMessage.getObject();
			
			if(msg != null)
				log.info("---->Getting message({}) from {}<----", msg, queueName);
           return msg;
		}
	@Override
	public void submitMessage(QueueEnum queueName, Serializable message) throws NamingException, JMSException {
			log.info("<----Submitting message({}) to {}---->", message.toString(), queueName);
			Destination destination=dMap.get(queueName);
			if (destination==null){
			 destination = (Destination)ic.lookup("/queue/"+queueName.toString());
			 dMap.put(queueName, destination);
			}
			MessageProducer producer = pMap.get(queueName);
			if(producer==null){
				producer = session.createProducer(destination);
			}
			
			ObjectMessage objMessage = session.createObjectMessage();
			objMessage.setObject(message);
			producer.send(objMessage, DeliveryMode.PERSISTENT, 0, MESSAGE_LIVE_TIME);
	}

    @Override
    public void commit() throws JMSException {
        session.commit();

    }

	@Override
	public void rollback() throws JMSException {
		session.rollback();
	}
}
